/* returns 
     *  FLB_HTTP_MORE - if we are waiting for more data to be received
     *  FLB_HTTP_CHUNK_AVAILABLE - if this is a chunked transfer and one or more chunks
     *                 have been received and it is not the end of the stream
     *  FLB_HTTP_OK - if we have collected all response data and no errors were thrown 
     *                (in chunked transfers this means we've received the end chunk 
     *                and any remaining data to process from the end of stream, will be
     *                contained in the response payload)
     *  FLB_HTTP_ERROR - for any error
     */

#define JSON_ARRAY_DELIM "\r\n"

static int process_http_chunk(struct k8s_events* ctx, struct flb_http_client *c, 
                size_t *bytes_consumed)
{
    int ret = 0;
    char *token = NULL;
    int root_type;
    size_t consumed = 0;
    char *buf_data = NULL;
    size_t buf_size;
    flb_sds_t payload;
    
    //we copy payload because tokenizer will overwrite when it finds end of string
    payload = flb_sds_create_len(c->resp.payload, c->resp.payload_size);

    token = strtok(payload, JSON_ARRAY_DELIM);
    while ( token != NULL && ret == 0 ) {
        ret = flb_pack_json(token, strlen(token), &buf_data, &buf_size, &root_type, &consumed);
        if (ret == -1) {
            //This is not actually an error as we may be in the middle of a received chunk
            flb_plg_debug(ctx->ins, "could not process payload, incomplete or bad formed JSON: %s", token);
        } else {
            *bytes_consumed += strlen(token) + 1;
            //ret = process_watched_event(ctx, buf_data, buf_size);
        }
    
        if (buf_data) {
            flb_free(buf_data);
        }
        buf_data = NULL;
        token = strtok(NULL, JSON_ARRAY_DELIM);
    }
    
    if (buf_data) {
        flb_free(buf_data);
    }
    flb_sds_destroy(payload);
    return ret;
}

static int k8s_events_collect(struct flb_input_instance *ins,
                              struct flb_config *config, void *in_context)
{
//... 
    //setup http client
    u_conn = flb_upstream_conn_get(ctx->upstream)
    c = flb_http_client(u_conn, FLB_HTTP_GET, "api/v1/events?watch=1",
                            NULL, 0, ctx->api_host, ctx->api_port, NULL, 0);
    flb_http_buffer_size(c, 0);
    flb_http_add_header(c, "User-Agent", 10, "Fluent-Bit", 10);
    if (ctx->auth_len > 0) {
        flb_http_add_header(c, "Authorization", 13, ctx->auth, ctx->auth_len);
    }

    // Watch will stream chunked json data, so we only send
    // the http request, then use flb_http_get_response_data
    // to attempt processing on available streamed data
    b_sent = 0;
    ret = flb_http_do_request(c, &b_sent);
    if (ret != 0) {
        flb_plg_error(ins, "http do error");
        goto exit;
    }

    ret = FLB_HTTP_MORE;
    size_t bytes_consumed = 0;
    while( ret == FLB_HTTP_MORE || ret == FLB_HTTP_CHUNK_AVAILABLE ) {        
        ret = flb_http_get_response_data(c, bytes_consumed);
        bytes_consumed = 0;
        if( ret == FLB_HTTP_CHUNK_AVAILABLE ) {
            process_http_chunk(ctx, c, &bytes_consumed);
            //todo: process return of process_http_chunk
        }
    }
    // TODO: process any remaining payload data on socket close (FLB_HTTP_OK) or process error (FLB_HTTP_ERROR)

[SERVICE]
    flush        1
    daemon       Off
    log_level    info
    http_server  Off
    http_listen  0.0.0.0
    http_port    2020

[INPUT]
    name          kubernetes_events
    tag           k8s_events
    kube_url      https://192.168.71.3:8443
    kube_ca_file  /Users/ryan/.minikube/ca.crt
    kube_token_file ./token

100% tests passed, 0 tests failed out of 133

Label Time Summary:
internal    =  98.57 sec*proc (57 tests)
runtime     = 1607.98 sec*proc (68 tests)

Total Test time (real) = 1743.18 sec